package yuekao5.dwd;

import com.alibaba.fastjson.JSON;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import yuekao5.util.KafkaUtil;

public class ReadOdsKafka {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //（DWD层事实表）编写Flink 流式程序，实时消费ODS层物流业务数据，过滤获取运单表（order_info）数据，按照运单状态（status）将运单数据划分不同类型数据，
        // 使用侧边流sideOutput方式，将不同状态运单数据存储Kafka不同Topic主题；（4分）
        DataStreamSource<String> streamSource = env.addSource(KafkaUtil.kafkaSource("ods-tms-topic"));

        SingleOutputStreamOperator<String> process = streamSource.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String s, ProcessFunction<String, String>.Context context, Collector<String> collector) throws Exception {
                String table = JSON.parseObject(s).getString("table");
                if ("order_info".equals(table)) {
                    collector.collect(s);
                }
            }
        });
//        process.print();
        OutputTag<String> tag60010 = new OutputTag<String>("60010") {
        };
        OutputTag<String> tag60020 = new OutputTag<String>("60020") {
        };
        OutputTag<String> tag60030 = new OutputTag<String>("60030") {
        };
        OutputTag<String> tag60040 = new OutputTag<String>("60040") {
        };
        OutputTag<String> tag60050 = new OutputTag<String>("60050") {
        };
        OutputTag<String> tag60060 = new OutputTag<String>("60060") {
        };
        OutputTag<String> tag60070 = new OutputTag<String>("60070") {
        };
        OutputTag<String> tag60080 = new OutputTag<String>("60080") {
        };
        OutputTag<String> tag60099 = new OutputTag<String>("60099") {
        };

        SingleOutputStreamOperator<String> process1 = process.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String s, ProcessFunction<String, String>.Context context, Collector<String> collector) throws Exception {
                String status = JSON.parseObject(s).getJSONObject("data").getString("status");
                if ("60010".equals(status)) {
                    context.output(tag60010, s);
                }
                if ("60020".equals(status)) {
                    context.output(tag60020, s);
                }
                if ("60030".equals(status)) {
                    context.output(tag60030, s);
                }
                if ("60040".equals(status)) {
                    context.output(tag60040, s);
                }
                if ("60050".equals(status)) {
                    context.output(tag60050, s);
                }
                if ("60060".equals(status)) {
                    context.output(tag60060, s);
                }
                if ("60070".equals(status)) {
                    context.output(tag60070, s);
                }
                if ("60080".equals(status)) {
                    context.output(tag60080, s);
                }
                if ("60099".equals(status)) {
                    context.output(tag60099, s);
                }
            }
        });
//        process1.getSideOutput(tag60010).print("60010");
//        process1.getSideOutput(tag60020).print("60020");
//        process1.getSideOutput(tag60030).print("60030");
//        process1.getSideOutput(tag60040).print("60040");
//        process1.getSideOutput(tag60050).print("60050");
//        process1.getSideOutput(tag60060).print("60060");
//        process1.getSideOutput(tag60070).print("60070");
//        process1.getSideOutput(tag60080).print("60080");
//        process1.getSideOutput(tag60099).print("60099");
        process1.getSideOutput(tag60010).addSink(KafkaUtil.kafkaSink("dwd-order-detail"));
        process1.getSideOutput(tag60020).addSink(KafkaUtil.kafkaSink("dwd-pay-detail"));
        process1.getSideOutput(tag60030).addSink(KafkaUtil.kafkaSink("tag60030"));
        process1.getSideOutput(tag60040).addSink(KafkaUtil.kafkaSink("dwd-receive-detail"));
        process1.getSideOutput(tag60050).addSink(KafkaUtil.kafkaSink("dwd-dispatch-detail"));
        process1.getSideOutput(tag60060).addSink(KafkaUtil.kafkaSink("dwd-bound-detail"));
        process1.getSideOutput(tag60070).addSink(KafkaUtil.kafkaSink("dwd-deliver-detail"));
        process1.getSideOutput(tag60080).addSink(KafkaUtil.kafkaSink("dwd-sign-detail"));
        process1.getSideOutput(tag60099).addSink(KafkaUtil.kafkaSink("dwd-cancel-detail"));
        env.execute();
    }
}
